package com.example.mq.mqserver.datacenter;

import com.example.mq.common.MqException;
import com.example.mq.mqserver.core.Binding;
import com.example.mq.mqserver.core.Exchange;
import com.example.mq.mqserver.core.MSGQueue;
import com.example.mq.mqserver.core.Message;

import java.io.IOException;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 使用这个类来统一管理内存中的数据,注意线程安全问题
 **/

public class MemoryDataCenter {
    // key是exchangeName，value是Exchange对象
    private ConcurrentHashMap<String, Exchange> exchangeMap = new ConcurrentHashMap<>();
    // key是queueName，value是MSGQueue对象
    private ConcurrentHashMap<String, MSGQueue> queueMap = new ConcurrentHashMap<>();
    // 第一个key是exchangeName，第二个key是queueName，value是Binding对象
    private ConcurrentHashMap<String, ConcurrentHashMap<String, Binding>> bindingsMap = new ConcurrentHashMap<>();
    // key是messageId，value是Message对象
    private ConcurrentHashMap<String, Message> messageMap = new ConcurrentHashMap<>();
    // 队列和消息的映射关系，key是queueName，value是Message对象的链表，此时的LinkedList是非线程安全的，之后需要进行处理
    private ConcurrentHashMap<String, LinkedList<Message>> queueMessageMap = new ConcurrentHashMap<>();
    // 第一个key是queueName，第二个key是messageId，value是Message对象
    private ConcurrentHashMap<String, ConcurrentHashMap<String, Message>> getQueueMessageWaitAck = new ConcurrentHashMap<>();

    // 封装交换机
    public void insertExchange(Exchange exchange) {
        exchangeMap.put(exchange.getName(), exchange);
        System.out.println("【MemoryDataCent】新交换机添加成功！exchangeName:"+exchange.getName());
    }

    public Exchange getExchange(String exchangeName) {
        return exchangeMap.get(exchangeName);
    }

    public void deleteExchange(String exchangeName) {
        exchangeMap.remove(exchangeName);
        System.out.println("【MemoryDataCent】交换机删除成功！exchangeName:"+exchangeName);
    }

    // 封装队列
    public void insertQueue(MSGQueue queue) {
        queueMap.put(queue.getName(), queue);
        System.out.println("【MemoryDataCent】新队列添加成功！queueName:"+queue.getName());
    }

    public MSGQueue getQueue(String queueName) {
        return queueMap.get(queueName);
    }

    public void deleteQueue(String queueName) {
        queueMap.remove(queueName);
        System.out.println("【MemoryDataCent】队列删除成功！queueName:"+queueName);
    }

    // 封装绑定操作
    public void insertBinding(Binding binding) throws MqException {
//        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(binding.getExchangeName());
//        if (bindingMap == null) {
//            bindingMap = new ConcurrentHashMap<>();
//            bindingsMap.put(binding.getExchangeName(), bindingMap);
//        }
        // 先查找exchangeName，如果对应的哈希表不存在就创建，bindingMap是一个ConcurrentHashMap，线程安全
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.computeIfAbsent(binding.getExchangeName(),
                k -> new ConcurrentHashMap<>());  // 等价于上面的代码

        synchronized (bindingMap) {
            // 再根据queueName查下，如果已存在就抛出异常，不存在才能插入
            if (bindingMap.containsKey(binding.getQueueName())) {
                throw new MqException("binding already exists！exchangeName:"+binding.getExchangeName()+
                        ", queueName:"+binding.getQueueName());
            }
            bindingMap.put(binding.getQueueName(), binding);
        }
        System.out.println("【MemoryDataCent】新绑定添加成功！exchangeName:"+binding.getExchangeName()+
                ", queueName:"+binding.getQueueName());
    }

    // 获取绑定 -- exchangeName和queueName获取指定的绑定，用exchangeName获取所有的绑定
    public Binding getBinding(String exchangeName, String queueName) {
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(exchangeName);
        if (bindingMap == null) {
            // 该交换机没有绑定任何队列
            return null;
        }
        return bindingMap.get(queueName);
    }

    public ConcurrentHashMap<String, Binding> getBindings(String exchangeName) {
        return bindingsMap.get(exchangeName);
    }

    // 删除绑定
    public void deleteBinding(Binding binding) throws MqException {
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(binding.getExchangeName());
        if (bindingMap == null) {
            // 该交换机没有绑定任何队列
            throw new MqException("binding not exists！exchangeName:"+binding.getExchangeName()+
                    ", queueName:"+binding.getQueueName());
        }
        bindingMap.remove(binding.getQueueName());
        System.out.println("【MemoryDataCent】绑定删除成功！exchangeName:"+binding.getExchangeName()+
                ", queueName:"+binding.getQueueName());
    }

    // 添加消息
    public void addMessage(Message message) {
        messageMap.put(message.getMessageId(), message);
        System.out.println("【MemoryDataCenter】新消息添加成功！messageId:"+message.getMessageId());
    }

    // 根据id查询消息
    public Message getMessage(String messageId) {
        return messageMap.get(messageId);
    }

    // 根据id删除消息
    public void removeMessage(String messageId) {
        messageMap.remove(messageId);
        System.out.println("【MemoryDataCenter】消息删除成功！messageId:"+messageId);
    }

    // 发送消息到指定队列
    public void sendMessage(MSGQueue queue, Message message) {
        // 把消息放到对于的队列数据结构中
        // 先根据队列名获取队列对应的链表
        // computeIfAbsent是线程安全的，不需要额外加锁，LinkedList不是线程安全的
        LinkedList<Message> messages = queueMessageMap.computeIfAbsent(queue.getName(), k -> new LinkedList<>());
//        LinkedList<Message> messageList = queueMessageMap.get(queue.getName());
//        if (messageList == null) {
//            // 该队列没有任何消息
//            messageList = new LinkedList<>();
//            queueMessageMap.put(queue.getName(), messageList);
//        }
        // 把消息放到链表中
        synchronized (messages) {
            messages.add(message);
        }
        // 这里把该消息也往消息中心插入，假设message已经在消息中心存在，重复插入也不会有问题
        // 因为相同的MessageId对应的message内容是相同的
        addMessage(message);
        System.out.println("【MemoryDataCenter】消息发送成功！messageId:"+message.getMessageId());
    }

    //  从队列中读取消息
    public Message pollMessage(String queueName) {
        // 先根据队列名获取队列对应的消息链表，没有找到说明队列中没有任何消息
        LinkedList<Message> messages = queueMessageMap.get(queueName);
        if (messages == null) {
            return null;
        }
        synchronized (messages) {
            // 如果没找到, 说明队列中没有任何消息.
            if (messages.isEmpty()) {
                return null;
            }
            // 链表中有元素, 就进行头删.
            Message currentMessage = messages.remove(0);
            System.out.println("[MemoryDataCenter] 消息从队列中取出! messageId=" + currentMessage.getMessageId());
            return currentMessage;
        }
    }

    // 获取指定队列中消息个数
    public int getMessageCount(String queueName) {
        LinkedList<Message> messages = queueMessageMap.get(queueName);
        if (messages == null) {
            return 0;
        }
        synchronized (messages) {
            return messages.size();
        }
    }

    // 添加未确认的消息
    public void addMessageWaitAck(String queueName, Message message) {
        ConcurrentHashMap<String, Message> messageMap = getQueueMessageWaitAck.computeIfAbsent(queueName,
                k -> new ConcurrentHashMap<>());
        messageMap.put(message.getMessageId(), message);
        System.out.println("【MemoryDataCenter】消息添加到未确认队列中！messageId:"+message.getMessageId());
    }

    // 删除未确认的消息
    public void removeMessageWaitAck(String queueName, String messageId) {
        ConcurrentHashMap<String, Message> messageMap = getQueueMessageWaitAck.get(queueName);
        if (messageMap == null) {
            return;
        }
        messageMap.remove(messageId);
        System.out.println("【MemoryDataCenter】消息从未确认队列中删除！messageId:"+messageId);
    }

    // 获取未确认的消息
    public Message getMessageWaitAck(String queueName, String messageId) {
        ConcurrentHashMap<String, Message> messageMap = getQueueMessageWaitAck.get(queueName);
        if (messageMap == null) {
            return null;
        }
        return messageMap.get(messageId);
    }

    // 把硬盘持久化存储的到各个维度的数据恢复到内存中
    public void recovery(DiskDataCenter diskDataCenter) throws IOException, ClassNotFoundException, MqException {
        // 0.清空之前的数据
        exchangeMap.clear();
        queueMap.clear();
        bindingsMap.clear();
        messageMap.clear();
        queueMessageMap.clear();
        // 1.恢复交换机
        List<Exchange> exchanges = diskDataCenter.selectAllExchanges();
        for (Exchange exchange : exchanges) {
            exchangeMap.put(exchange.getName(), exchange);
        }
        // 2.恢复队列
        List<MSGQueue> queues = diskDataCenter.selectAllQueues();
        for (MSGQueue queue : queues) {
            queueMap.put(queue.getName(), queue);
        }
        // 3.恢复绑定
        List<Binding> bindings = diskDataCenter.selectAllBindings();
        for (Binding binding : bindings) {
            ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.computeIfAbsent(binding.getExchangeName(),
                    k -> new ConcurrentHashMap<>());
            bindingMap.put(binding.getQueueName(), binding);
        }
        // 4.恢复消息
        //   遍历所有队列，根据队列名字获取所有消息
        for (MSGQueue queue : queues) {
            LinkedList<Message> messages = diskDataCenter.loadAllMessageFromQueue(queue.getName());
            queueMessageMap.put(queue.getName(), messages);
            // 把消息放到消息中心
            for (Message message : messages) {
                messageMap.put(message.getMessageId(), message);
            }
        }
        // 针对未确认的消息，不需要从硬盘恢复，一旦重启，未确认的消息恢复为未被取走的消息
        // 这个消息在硬盘上存储的时候，就当作未被取走的消息
    }
}
